package com.sk.govern.client;


import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.*;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.List;
import java.util.Objects;

/**
 * description：使用curator作为zk的client连接工具
 * [功能点] 对znode的crud、初始化client（包含重试策略）
 *
 * @author 孙文斌
 * @Date: 2021/1/31 13:07
 */

@Data
@Slf4j
@Component
public class CuratorClient {
    @Value("${zk.curator.connectUrls}")
    private String connectUrls;
    @Value("${zk.curator.retryTime}")
    private String retryTime;
    @Value("${zk.curator.connectionTimeOut}")
    private String connectionTimeOut;
    @Value("${zk.curator.sessionTimeOut}")
    private String sessionTimeOut;
    @Value("${zk.curator.baseSleepTime}")
    private String baseSleepTime;
    private CuratorFramework client;


    @PostConstruct
    private CuratorFramework init() {
        try {
            client = CuratorFrameworkFactory.builder()
                    .connectString(connectUrls)
                    .connectionTimeoutMs(Integer.valueOf(connectionTimeOut))
                    .sessionTimeoutMs(Integer.valueOf(sessionTimeOut))
                    .retryPolicy(new ExponentialBackoffRetry(Integer.valueOf(baseSleepTime), Integer.valueOf(retryTime)))
                    .build();
            client.start();
//            listenerChild("/com.sk.common.level2.rpcInterface.CalculateRpcInterface/provider/add-1415560896");
            register();
        } catch (Exception e) {
            e.printStackTrace();
            log.error("init zookeeper client fail,param is " + this.toString());
        }
        return client;
    }
    public void register() {
//        String interfacePath = "/" + "com.sk.common.level2.rpcInterface.CalculateRpcInterface";
        try {
//            if (this.client.checkExists().forPath(interfacePath) == null) {
//                // 创建 服务的永久节点
//                this.client.create()
//                        .creatingParentsIfNeeded()
//                        .withMode(CreateMode.PERSISTENT)
//                        .forPath(interfacePath);
//            }
//            String methodPath = String.format("%s/%s/%s/%s", interfacePath, "consumer", "add-1415560896", "192.168.59.3:18081");
//            log.info("zk path: [" + methodPath + "]");
//            String s1 = this.client.create()
//                    .creatingParentsIfNeeded()
//                    .withMode(CreateMode.EPHEMERAL)//最末尾的是临时节点
//                    .forPath(methodPath, "0".getBytes());
            //
//            String s = this.client.create().creatingParentsIfNeeded()
//                    .withMode(CreateMode.EPHEMERAL).forPath("/vincent/sk01", "0".getBytes());
            String s7 = this.client.create().creatingParentsIfNeeded()
                    .withMode(CreateMode.EPHEMERAL).forPath("/vincent/sk", "0".getBytes());
            System.out.println("打印："+this.client.checkExists().forPath("/vincent/sk"));
            if (this.client.checkExists().forPath("/vincent/sk") != null) {
                Void aVoid = this.client.delete().forPath("/vincent/sk");
                System.out.println("删除成功！");
            }
//
//            String s3 = this.client.create().creatingParentsIfNeeded()
//                    .withMode(CreateMode.EPHEMERAL).forPath("/vincent/sk0", "1".getBytes());
//            System.out.println(s3);
            // 创建临时节点，节点包含了服务提供段的信息

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void listenerChild(String path) {

        try {
            PathChildrenCache pathChildrenCache = new PathChildrenCache(this.client, path, false);//只能监听子节点，不能递归监听子节点下的子节点
            //构造一个子节点缓存监听器PathChildrenCacheListener实例
            PathChildrenCacheListener listener = new PathChildrenCacheListener() {
                @Override
                public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent event) throws Exception {
                    try {
                        List<ChildData> childDataList = pathChildrenCache.getCurrentData();
                        System.out.println("event数据为："+event.toString());
                        System.out.println("有动静,类型为："+event.getType()+"数据为："+childDataList.toString());
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            };
            pathChildrenCache.getListenable().addListener(listener);
            //NORMAL——异步初始化cache
            //BUILD_INITIAL_CACHE——同步初始化cache
            //POST_INITIALIZED_EVENT——异步初始化cache，并触发完成事件
            pathChildrenCache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
            //首次拉取provide信息
            //todo:暂时不考虑更新和首次拉取的次序问题
//            List<ChildData> currentData = pathChildrenCache.getCurrentData();
//            System.out.println("当前的子节点是："+currentData.toString());
        } catch (Exception e) {
            e.printStackTrace();
            log.error("使用缓存监听模式(Path Cache)对consumer监听失败，e+" + e);
        }
    }

    //增加znode（重试策略）
    /**
     * @param path
     * @param value
     * @Description: 创建节点
     * @Date: 2021/1/31 13:07
     */
    public String createNode(String path, String value) throws Exception {
        if (null == client) {
            throw new RuntimeException("there is not connect to zkServer...");
        }
        String node = client
                .create()
                .creatingParentsIfNeeded()
                .withMode(CreateMode.EPHEMERAL_SEQUENTIAL) // 临时顺序节点
                .forPath(path, value.getBytes());

        log.info("create node : {}", node);
        return node;
    }
    /**
     * @param path
     * @Description: 删除节点信息
     * @Date: 2020-08-22 16:01
     */
    public void deleteNode(String path) throws Exception {
        if (null == client) {
            throw new RuntimeException("there is not connect to zkServer...");
        }
        client
                .delete()
                .deletingChildrenIfNeeded()
                .forPath(path);
        log.info("{} is deleted ", path);
    }


    /**
     * @param path
     * @Description: 获取节点存储的值
     * @Date: 2020-08-22 16:11
     */
    public String getNodeData(String path) throws Exception {
        if (null == client) {
            throw new RuntimeException("there is not connect to zkServer...");
        }
        Stat stat = new Stat();
        byte[] bytes = client.getData().storingStatIn(stat).forPath(path);
        log.info("{} data is : {}", path, new String(bytes));
        log.info("current stat version is {}, createTime is {}", stat.getVersion(), stat.getCtime());
        return new String(bytes);
    }


    /**
     * @Description: 设置节点 数据
     * @param path
     * @param value
     * @Date: 2020-08-22 16:17
     */
    public void setNodeData(String path,String value) throws Exception {
        if (null == client) {
            throw new RuntimeException("there is not connect to zkServer...");
        }
        Stat stat = client.checkExists().forPath(path);
        if(null == stat){
            log.info(String.format("{} Znode is not exists",path));
            throw new RuntimeException(String.format("{} Znode is not exists",path));
        }
        String nodeData = getNodeData(path);
        client.setData().withVersion(stat.getVersion()).forPath(path, value.getBytes());
        log.info("{} Znode data is set. old vaule is {}, new data is {}",path,nodeData,value);
    }


    /**
     * @Description: 创建 给定节点的监听事件  监听一个节点的更新和创建事件(不包括删除)
     * @param path
     * @Date: 2020-08-22 16:47
     */
    public void addWatcherWithNodeCache(String path) throws Exception {
        if (null == client) {
            throw new RuntimeException("there is not connect to zkServer...");
        }
        // dataIsCompressed if true, data in the path is compressed
        NodeCache nodeCache = new NodeCache(client, path,false);
        NodeCacheListener listener = () -> {
            ChildData currentData = nodeCache.getCurrentData();
            log.info("{} Znode data is chagnge,new data is ---  {}", currentData.getPath(), new String(currentData.getData()));
        };
        nodeCache.getListenable().addListener(listener);
        nodeCache.start();
    }


    /**
     * @Description: 监听给定节点下的子节点的创建、删除、更新
     * @param path 给定节点
     * @Date: 2020-08-22 17:14
     */
    public void addWatcherWithChildCache(String path) throws Exception {
        if (null == client) {
            throw new RuntimeException("there is not connect to zkServer...");
        }
        //cacheData if true, node contents are cached in addition to the stat
        PathChildrenCache pathChildrenCache = new PathChildrenCache(client,path,false);
        PathChildrenCacheListener listener = (client, event) -> {
            log.info("event path is --{} ,event type is {}" , event.getData().getPath(), event.getType());
        };
        pathChildrenCache.getListenable().addListener(listener);
        // StartMode : NORMAL  BUILD_INITIAL_CACHE  POST_INITIALIZED_EVENT
        pathChildrenCache.start(PathChildrenCache.StartMode.NORMAL);
    }

    /**
     * @Description: 监听 给定节点的创建、更新（不包括删除） 以及 该节点下的子节点的创建、删除、更新动作。
     * @param path
     * @Date: 2020-08-22 17:35
     */
    public void addWatcherWithTreeCache(String path) throws Exception {
        if (null == client) {
            throw new RuntimeException("there is not connect to zkServer...");
        }
        TreeCache treeCache = new TreeCache(client, path);
        TreeCacheListener listener = (client, event) -> {
            log.info("节点路径 --{} ,节点事件类型: {} , 节点值为: {}" , Objects.nonNull(event.getData()) ? event.getData().getPath() : "无数据", event.getType());
        };
        treeCache.getListenable().addListener(listener);
        treeCache.start();
    }
    /**
     * @param
     * @Description: 关闭连接
     * @Date: 2021/1/31 13:07
     */
    public void closeConnection() {
        if (null != client) {
            client.close();
        }
    }
}
